package yhb.component

import org.apache.hadoop.hbase.io.compress.Compression
import org.apache.hadoop.hbase.{TableName, HColumnDescriptor, HTableDescriptor, HBaseConfiguration}
import org.apache.hadoop.hbase.client.HBaseAdmin

/**
 * Created by root on 15-9-6.
 */
object HandleHbase {

  val quorum = "slave01,slave02,slave03,slave04,slave05"
  val conf = HBaseConfiguration.create()

  {
    conf.set("hbase.zookeeper.property.clientPort", "2222")
    conf.set("hbase.zookeeper.quorum", quorum)
    conf.set("hbase.master", "master1:60000")
  }

  def newConfiguration()={
    val newConf = HBaseConfiguration.create()
    newConf.set("hbase.zookeeper.property.clientPort", "2222")
    newConf.set("hbase.zookeeper.quorum", quorum)
    newConf.set("hbase.master", "master1:60000")
    newConf
  }

//  def getTable(tableName:String)={
//    val newConf = HBaseConfiguration.create()
//    newConf.set("hbase.zookeeper.property.clientPort", "2222")
//    newConf.set("hbase.zookeeper.quorum", quorum)
//    newConf.set("hbase.master", "master1:60000")
//    val connection = ConnectionFactory.createConnection(newConf)
//    connection.getTable(TableName.valueOf(tableName))
//  }

  /**
   * 这个方法有bug，集群用不了
   * @param tableName
   * @param columnFamilys
   */
  def createTable(tableName:String,columnFamilys:String*){
    val admin = new HBaseAdmin(conf)
    if(admin.tableExists(tableName)){
      admin.disableTable(tableName)
      admin.deleteTable(tableName)
    }
    val tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName))
//    for(family<-columnFamilys) tableDescriptor.addFamily(new HColumnDescriptor(family.getBytes))
//    下面这句是尝试优化,使用压缩存储
    for(family<-columnFamilys) tableDescriptor.addFamily(new HColumnDescriptor(family.getBytes).setCompressionType(Compression.Algorithm.GZ))
    admin.createTable(tableDescriptor)
    //下面的代码是hbase 1.0+版本使用
//    val connection = ConnectionFactory.createConnection(conf)
//    val admin = connection.getAdmin
//    val table = new HTableDescriptor(TableName.valueOf(tableName))
//    if(admin.tableExists(table.getTableName)){
//      admin.disableTable(table.getTableName())
//      admin.deleteTable(table.getTableName())
//    }
//    for(family<-columnFamilys) table.addFamily(new HColumnDescriptor(family).setCompressionType(Algorithm.SNAPPY))
//    admin.createTable(table)
  }

}
